/******************************************************************************/
/* @ All Rights Reserved Copyright (C) 2014 broada,Ltd.                       */
/*                                                                            */
/*  NAME      = kafka_core.c                                                  */
/*  NOTE      = KAFKA协议解析                                                 */
/*  DATE      = 2015/03/18 by zhengxd                                         */
/******************************************************************************/
#include <stdio.h>
#include <string.h>
#include <netinet/in.h>

#include "frservice_comm.h"
#include "kafka_core.h"
#include "kafka_comm.h"
#include "kafka_request.h"
#include "kafka_response.h"
#include "config.h"

#define KAFKA_MAX_BUFF_SIZE      1024 * 1024
#define KAFKA_MAX_PACKETS        200

extern void *g_p_apm_context;

/*******************************************************************************
*  FUNC     :  设置方向
*  ARGS     :  p_st_session(KAFKA私有session)
*  RTN      :  FR_SUCC       (normal)
*              FR_FAIL       (error)
*  NOTE     :  p_st_session必须不为NULL
*******************************************************************************/
static int __kafka_set_cs_srv(Conntrack *p_st_conn
                               , Apm_packet *p_st_packet
                               , int i_version)
{
    int i_ret = FR_SUCC;

    /* 设置方向 */
    if(frct_get_direct_known(p_st_conn) == FR_DIRECT_UNKNOWN)
    {
        i_ret = frcap_set_cs_srv(p_st_conn, p_st_packet->dst_ip
                                , p_st_packet->dst_port);
        if (i_ret != FR_SUCC)
        {
            return FR_FAIL;
        }
    }
    frservice_add(g_p_apm_context, MODULE_L7PROTO, p_st_packet->dst_ip, p_st_packet->dst_port, &i_version, sizeof(int));
    return FR_SUCC;
}

/*******************************************************************************
*  FUNC     :  初始化私有KAFKA session
*  ARGS     :  p_st_session(KAFKA私有session)
*  RTN      :  FR_SUCC       (normal)
*              FR_FAIL       (error)
*  NOTE     :  p_st_session必须不为NULL
*******************************************************************************/
static int __kafka_init_private(Conntrack *p_st_conn
                                , Apm_packet *p_st_packet
                                , int i_version)
{
    L7_session *p_st_session = NULL;
    L7_fragment_kafka *p_st_fragment = NULL;               /* 分片包相关数据 */
    L7session_kafka_private *p_st_private_session = NULL;  /* KAFKA私有session */

    /* 到Conntrack中取得session信息 */
    p_st_session = frct_get_session(p_st_conn);
    if (p_st_session == NULL)
    {
        return FR_FAIL;
    }
    /* 设置session ID */
    fruuid(p_st_session->arr_session_id);

    /* 申请整个私有session结构的内存 */
    p_st_private_session = (L7session_kafka_private *)frmalloc(sizeof(L7session_kafka_private));
    if (p_st_private_session == NULL)
    {
        goto error;
    }
    memset(p_st_private_session, 0x00, sizeof(L7session_kafka_private));

    /* 申请分片包信息结构内存 */
    p_st_fragment = (L7_fragment_kafka *)frmalloc(sizeof(L7_fragment_kafka));
    if (p_st_fragment == NULL)
    {
        goto error;
    }
    memset(p_st_fragment, 0x00, sizeof(L7_fragment_kafka));

    /* 设置初始化结构 */
    p_st_private_session->i_version = i_version;
    p_st_private_session->p_st_fragment = p_st_fragment;
    p_st_session->p_session_private = p_st_private_session;
    return FR_SUCC;

error:
    kafka_free((char **)(&p_st_private_session));
    kafka_free((char **)(&p_st_fragment));
    return FR_FAIL;
}

/*******************************************************************************
*  FUNC     :  初始化私有KAFKA session
*  ARGS     :  p_st_private(私有session数据)
*  RTN      :  FR_SUCC       (normal)
*              FR_FAIL       (error)
*  NOTE     :  p_st_private必须不是NULL
*******************************************************************************/
static void __kafka_destroy_curr_info(L7session_kafka_private *p_st_private)
{
    if (p_st_private->p_st_req)
    {
        kafka_destroy_request(&p_st_private->p_st_req);
    }
    if (p_st_private->p_st_resp)
    {
        kafka_destroy_response(&p_st_private->p_st_resp);
    }
    kafka_init_fragment(p_st_private->p_st_fragment);

    return;
}

/*******************************************************************************
*  FUNC     :  取得完整的packet包
*  ARGS     :  p_st_private(私有session数据)
*           :  p_data(当前数据包数据)
*           :  ui_len(当前数据包数据长度)
*           :  p_st_timestamp(当前数据包发生的时间点)
*  RTN      :  APM_L7_ANALYZE_FAIL      (FAIL)
*              APM_L7_ANALYZE_TODO      (More data)
*              APM_L7_ANALYZE_SUCC      (SUCC)
*  NOTE     :  解析失败后需要释放p_st_private里的request/response/fragment信息
*******************************************************************************/
static int __kafka_set_analysis_data(Conntrack *p_st_conn
                                    , Apm_packet *p_st_packet
                                    , L7session_kafka_private *p_st_private)
{
    int i_ret = FR_SUCC;
    int i_req_resp = 0;
    uint32_t u32_total_len = 0;
    unsigned int ui_len = p_st_packet->ui_payload_len;
    char *p_data = p_st_packet->p_payload;
    L7_fragment_kafka *p_st_fragment = p_st_private->p_st_fragment;

    /* 无数据，跳过该包 */
    if (p_st_packet->ui_payload_len == 0)
    {
        return APM_L7_ANALYZE_TODO;
    }
    /* 取得方向 */
    i_req_resp = frcap_get_cs_direction(p_st_conn, p_st_packet->src_ip
                                       , p_st_packet->src_port);
    if (p_st_fragment->p_st_data == NULL)
    {
        p_st_fragment->p_st_data = bytebuff_creat();
        if (p_st_fragment->p_st_data == NULL)
        {
            return APM_L7_ANALYZE_FAIL;
        }
        i_ret = bytebuff_add(p_st_fragment->p_st_data, p_data, ui_len);
        if (i_ret != FR_SUCC)
        {
            return APM_L7_ANALYZE_FAIL;
        }

        /* 取得长度 */
        i_ret = bytebuff_get_int(p_st_fragment->p_st_data, 4, &u32_total_len, BYTE_BIG_ENDIAN);
        if (i_ret != FR_SUCC || u32_total_len + 4 < ui_len)
        {
            return APM_L7_ANALYZE_FAIL;
        }

        if (u32_total_len >= KAFKA_MAX_BUFF_SIZE)
        {
            frdbg("too large size!\n")
            return APM_L7_ANALYZE_FAIL;
        }

        u32_total_len += 4;
        p_st_fragment->ui_total_len = u32_total_len;
        p_st_fragment->st_start_timestamp = p_st_packet->st_timestamp;
        p_st_fragment->i_req_resp = i_req_resp;
        /* 没有其他分片包 */
        if (u32_total_len == ui_len)
        {
            p_st_fragment->st_end_timestamp = p_st_packet->st_timestamp;
            return APM_L7_ANALYZE_SUCC;
        }
        else
        {
            p_st_fragment->ui_remain_len = u32_total_len - ui_len;
            return APM_L7_ANALYZE_TODO;
        }
    }
    else
    {
        /* 前后两次方向不一致, 消除上次信息 */
        if (p_st_fragment->i_req_resp != i_req_resp)
        {
            kafka_init_fragment(p_st_fragment);
            i_ret = __kafka_set_analysis_data(p_st_conn, p_st_packet, p_st_private);
            return i_ret;
        }
        else
        {
            if (p_st_fragment->ui_remain_len < ui_len)
            {
                kafka_init_fragment(p_st_fragment);
                i_ret = __kafka_set_analysis_data(p_st_conn, p_st_packet, p_st_private);
                return i_ret;
            }
            else if (p_st_fragment->ui_remain_len == ui_len)
            {
                p_st_fragment->st_end_timestamp = p_st_packet->st_timestamp;
                p_st_fragment->ui_remain_len = 0;
                i_ret = bytebuff_add(p_st_fragment->p_st_data, p_data, ui_len);
                if (i_ret != FR_SUCC)
                {
                    return APM_L7_ANALYZE_FAIL;
                }
                return APM_L7_ANALYZE_SUCC;

            }
            else
            {
                if (frlist_size(p_st_fragment->p_st_data->p_lst_bytebuffer) >= KAFKA_MAX_PACKETS)
                {
                    frdbg("too many packets!\n")
                    return APM_L7_ANALYZE_FAIL;
                }
                p_st_fragment->ui_remain_len -= ui_len;
                i_ret = bytebuff_add(p_st_fragment->p_st_data, p_data, ui_len);
                if (i_ret != FR_SUCC)
                {
                    return APM_L7_ANALYZE_FAIL;
                }
                return APM_L7_ANALYZE_TODO;
            }
        }
    }
}

/*******************************************************************************
*  FUNC     :  判断当前数据链路是否是KAFKA协议
*  ARGS     :  p_st_conn(入力链接信息)
*           :  p_st_packet(入力数据包)
*  RTN      :  APM_L7_IS_NO   (NO)
*              APM_L7_IS_YES  (YES)
*  NOTE     :
*******************************************************************************/
int kafka_detect(Conntrack *p_st_conn, Apm_packet *p_st_packet)
{
    int i_ret = FR_SUCC;
    uint32_t u32_len = 0;
    uint16_t u16_len = 0;
    uint16_t u16_req_type = 0;
    uint16_t u16_topic_len = 0;
    uint16_t u16_topic_count = 0;
    int i_version = 0;
    unsigned int ui_len_tmp = 0;
    unsigned int ui_len = p_st_packet->ui_payload_len;  /* packet数据长度 */
    char *p_data = p_st_packet->p_payload;              /* packet数据内容 */
    Frbyte_Packet *p_st_data = NULL;
    char arr_arr[64] = {0};

    if(p_st_packet == NULL || p_st_packet->p_payload == NULL || p_st_packet->ui_payload_len == 0 || p_st_conn==NULL)
    {
        return APM_L7_IS_NO;
    }

    if (ui_len < 4)
    {
        goto unknown;
    }
    /* 创建bytebuff */
    p_st_data = bytebuff_creat();
    if (p_st_data == NULL)
    {
        goto unknown;
    }
    i_ret = bytebuff_add(p_st_data, p_data, ui_len);
    if (i_ret != FR_SUCC)
    {
        goto unknown;
    }

    /* 取得长度 */
    i_ret = bytebuff_get_int(p_st_data, 4, &u32_len, BYTE_BIG_ENDIAN);
    if (i_ret != FR_SUCC || u32_len != ui_len - 4)
    {
        goto unknown;
    }

    /* get request type */
    i_ret = bytebuff_get_short(p_st_data, 2, &u16_req_type, BYTE_BIG_ENDIAN);
    if (i_ret != FR_SUCC)
    {
        goto unknown;
    }

    /* 0.7以下type判定 */
    switch (u16_req_type)
    {
        case KAFKA_REQ_TP_PRODUCE:
            /* get topic length */
            i_ret = bytebuff_get_short(p_st_data, 2, &u16_topic_len, BYTE_BIG_ENDIAN);
            if (i_ret != FR_SUCC)
            {
                break;
            }
            /* skip topic and partition */
            i_ret = bytebuff_skip(p_st_data, u16_topic_len + 4);
            if (i_ret != FR_SUCC)
            {
                break;
            }
            /* get message length */
            i_ret = bytebuff_get_int(p_st_data, 4, &u32_len, BYTE_BIG_ENDIAN);
            if (i_ret != FR_SUCC || ui_len != 4 + 2 + 2 + u16_topic_len + 4 + 4 + u32_len)
            {
                break;
            }
            i_ret = __kafka_set_cs_srv(p_st_conn, p_st_packet, KAFKA_VERSION_7);
            bytebuff_free(&p_st_data);
            if (i_ret != FR_SUCC)
            {
                return APM_L7_IS_NO;
            }
            return APM_L7_IS_YES;

        case KAFKA_REQ_TP_FETCH:
        case KAFKA_REQ_TP_OFFSETS:
            /* get topic length */
            i_ret = bytebuff_get_short(p_st_data, 2, &u16_topic_len, BYTE_BIG_ENDIAN);
            if (i_ret != FR_SUCC || ui_len != (unsigned int)(4 + 2 + 2 + u16_topic_len + 4 + 8 + 4))
            {
                break;
            }
            i_ret = __kafka_set_cs_srv(p_st_conn, p_st_packet, KAFKA_VERSION_7);
            bytebuff_free(&p_st_data);
            if (i_ret != FR_SUCC)
            {
                return APM_L7_IS_NO;
            }
            return APM_L7_IS_YES;

        case KAFKA_REQ_TP_MULTIFETCH:
            /* get topic length */
            i_ret = bytebuff_get_short(p_st_data, 2, &u16_topic_count, BYTE_BIG_ENDIAN);
            ui_len_tmp = ui_len - 4 - 2;
            while (u16_topic_count--)
            {
                /* get topic length */
                i_ret = bytebuff_get_short(p_st_data, 2, &u16_topic_len, BYTE_BIG_ENDIAN);
                if (i_ret != FR_SUCC || ui_len_tmp < (unsigned int)(2 + u16_topic_len + 4 + 8 + 4))
                {
                    break;
                }
                i_ret = bytebuff_skip(p_st_data, u16_topic_len + 4 + 8 + 4);
                if (i_ret != FR_SUCC)
                {
                    break;
                }
                ui_len_tmp -= 2 + u16_topic_len + 4 + 8 + 4;
            }

            if (ui_len_tmp)
            {
                break;
            }
            i_ret = __kafka_set_cs_srv(p_st_conn, p_st_packet, KAFKA_VERSION_7);
            bytebuff_free(&p_st_data);
            if (i_ret != FR_SUCC)
            {
                return APM_L7_IS_NO;
            }
            return APM_L7_IS_YES;

        case KAFKA_REQ_TP_MULTIPRODUCE:
            /* get topic length */
            i_ret = bytebuff_get_short(p_st_data, 2, &u16_topic_count, BYTE_BIG_ENDIAN);
            ui_len_tmp = ui_len - 4 - 2;
            while (u16_topic_count--)
            {
                /* get topic length */
                i_ret = bytebuff_get_short(p_st_data, 2, &u16_topic_len, BYTE_BIG_ENDIAN);
                if (i_ret != FR_SUCC || ui_len_tmp < (unsigned int)(2 + u16_topic_len + 4))
                {
                    break;
                }
                i_ret = bytebuff_skip(p_st_data, u16_topic_len + 4);
                if (i_ret != FR_SUCC)
                {
                    break;
                }
                /* get message length */
                i_ret = bytebuff_get_int(p_st_data, 4, &u32_len, BYTE_BIG_ENDIAN);
                if (i_ret != FR_SUCC || ui_len < (unsigned int)(2 + u16_topic_len + 4 + 4 + u32_len))
                {
                    break;
                }
                i_ret = bytebuff_skip(p_st_data, u32_len);
                if (i_ret != FR_SUCC)
                {
                    break;
                }
                ui_len_tmp -= 2 + u16_topic_len + 4 + 4 + u32_len;
            }
            if (ui_len_tmp)
            {
                break;
            }
            i_ret = __kafka_set_cs_srv(p_st_conn, p_st_packet, KAFKA_VERSION_7);
            bytebuff_free(&p_st_data);
            if (i_ret != FR_SUCC)
            {
                return APM_L7_IS_NO;
            }
            return APM_L7_IS_YES;

        default:
            break;
    }

    /* set position */
    bytebuff_setPosition(p_st_data, 4 + 2);
    /* skip ApiVersion CorrelationId */
    i_ret = bytebuff_skip(p_st_data, 2 + 4);
    if (i_ret != FR_SUCC)
    {
        goto unknown;
    }
    /* skip ClientId */
    i_ret = bytebuff_get_short(p_st_data, 2, &u16_len, BYTE_BIG_ENDIAN);
    if (i_ret != FR_SUCC)
    {
        goto unknown;
    }
    if (u16_len != 0 && u16_len != 0xFFFF)
    {
        i_ret = bytebuff_skip(p_st_data, u16_len);
        if (i_ret != FR_SUCC)
        {
            goto unknown;
        }
    }

    /* type判定 */
    switch (u16_req_type)
    {
        case KAFKA_REQ_TP_PRODUCE:
            /* skip RequiredAcks Timeout */
            i_ret = bytebuff_skip(p_st_data, 2 + 4);
            if (i_ret != FR_SUCC)
            {
                goto unknown;
            }
            strncpy(arr_arr, "5[36]", 64);
            i_ret = kafka_skip_arr(p_st_data, strlen(arr_arr), arr_arr);
            break;

        case KAFKA_REQ_TP_FETCH:
            /* skip ReplicaId MaxWaitTime MinBytes */
            i_ret = bytebuff_skip(p_st_data, 4 + 4 + 4);
            if (i_ret != FR_SUCC)
            {
                goto unknown;
            }
            strncpy(arr_arr, "5[343]", 64);
            i_ret = kafka_skip_arr(p_st_data, strlen(arr_arr), arr_arr);
            break;

        case KAFKA_REQ_TP_OFFSET:
            /* skip ReplicaId */
            i_ret = bytebuff_skip(p_st_data, 4);
            if (i_ret != FR_SUCC)
            {
                goto unknown;
            }
            strncpy(arr_arr, "5[343]", 64);
            i_ret = kafka_skip_arr(p_st_data, strlen(arr_arr), arr_arr);
            break;

        case KAFKA_REQ_TP_METADATA:
            strncpy(arr_arr, "5", 64);
            i_ret = kafka_skip_arr(p_st_data, strlen(arr_arr), arr_arr);
            break;

        case KAFKA_REQ_TP_OFFSET_COMMIT:
            /* skip ConsumerGroupId */
            i_ret = kafka_skip_string(p_st_data);
            if (i_ret != FR_SUCC)
            {
                goto unknown;
            }
            strncpy(arr_arr, "5[345]", 64);
            i_ret = kafka_skip_arr(p_st_data, strlen(arr_arr), arr_arr);
            break;

        case KAFKA_REQ_TP_OFFSET_FETCH:
            /* skip ConsumerGroup */
            i_ret = kafka_skip_string(p_st_data);
            if (i_ret != FR_SUCC)
            {
                goto unknown;
            }
            strncpy(arr_arr, "5[3]", 64);
            i_ret = kafka_skip_arr(p_st_data, strlen(arr_arr), arr_arr);
            break;

        default:
            goto unknown;
    }

    if (i_ret != APM_L7_ANALYZE_SUCC || bytebuff_remaining(p_st_data))
    {
        goto unknown;
    }
    bytebuff_free(&p_st_data);

    i_ret = __kafka_set_cs_srv(p_st_conn, p_st_packet, KAFKA_VERSION_8);
    if (i_ret != FR_SUCC)
    {
        return APM_L7_IS_NO;
    }
    return APM_L7_IS_YES;

unknown:
    bytebuff_free(&p_st_data);
    if (frct_get_direct_known(p_st_conn) == FR_DIRECT_UNKNOWN)
    {
        i_ret = frservice_get(g_p_apm_context, MODULE_L7PROTO, p_st_packet->dst_ip, p_st_packet->dst_port, &i_version, sizeof(int), NULL);
        if (i_ret == FR_SUCC)
        {
            frcap_set_cs_srv(p_st_conn, p_st_packet->dst_ip, p_st_packet->dst_port);
            return APM_L7_IS_YES;
        }
        i_ret = frservice_get(g_p_apm_context, MODULE_L7PROTO, p_st_packet->src_ip, p_st_packet->src_port, &i_version, sizeof(int), NULL);
        if (i_ret == FR_SUCC)
        {
            frcap_set_cs_srv(p_st_conn, p_st_packet->src_ip, p_st_packet->src_port);
            return APM_L7_IS_YES;
        }
    }
    return APM_L7_IS_NO;
}

/*******************************************************************************
*  FUNC     :  KAFKA的packet解析
*  ARGS     :  p_st_conn(入力链接信息)
            :  p_st_packet(入力packet信息)
            :  p_st_hash_report(保存Frreport_data)
*  RTN      :  参照module.h
*  NOTE     :
*******************************************************************************/
int kafka_analysis(Conntrack *p_st_conn
                 , Apm_packet *p_st_packet
                 , Frhash_header *p_st_hash_report)
{
    int i_ret = FR_SUCC;
    int i_sub_ret = FR_SUCC;
    int i_version = 0;
    L7_session *p_st_session = NULL;              /* L7session */
    L7session_kafka_private *p_st_private = NULL;   /* KAFKA的私有数据 */

    /* 到Conntrack中取得session信息 */
    p_st_session = frct_get_session(p_st_conn);
    if (p_st_session == NULL)
    {
        return APM_L7_ANALYZE_FAIL;
    }

    if (p_st_session->p_session_private == NULL)
    {
        i_ret = frservice_get(g_p_apm_context, MODULE_L7PROTO, p_st_packet->dst_ip, p_st_packet->dst_port, &i_version, sizeof(int), &i_sub_ret);
        if (i_ret == FR_SUCC && i_sub_ret == FR_SUCC)
        {
            i_ret = __kafka_init_private(p_st_conn, p_st_packet, i_version);
            if (i_ret != FR_SUCC)
            {
                return APM_L7_ANALYZE_FAIL;
            }
        }
        else
        {
            i_ret = frservice_get(g_p_apm_context, MODULE_L7PROTO, p_st_packet->src_ip, p_st_packet->src_port, &i_version, sizeof(int), &i_sub_ret);
            if (i_ret == FR_SUCC && i_sub_ret == FR_SUCC)
            {
                i_ret = __kafka_init_private(p_st_conn, p_st_packet, i_version);
                if (i_ret != FR_SUCC)
                {
                    return APM_L7_ANALYZE_FAIL;
                }
            }
            else
            {
                i_ret = kafka_detect(p_st_conn, p_st_packet);
                if (i_ret != APM_L7_IS_YES)
                {
                    return APM_L7_ANALYZE_FAIL;
                }
                i_ret = frservice_get(g_p_apm_context, MODULE_L7PROTO, p_st_packet->dst_ip, p_st_packet->dst_port, &i_version, sizeof(int), &i_sub_ret);
                if (i_ret == FR_SUCC && i_sub_ret == FR_SUCC)
                {
                    i_ret = __kafka_init_private(p_st_conn, p_st_packet, i_version);
                    if (i_ret != FR_SUCC)
                    {
                        return APM_L7_ANALYZE_FAIL;
                    }
                }
                else
                {
                    return APM_L7_ANALYZE_FAIL;
                }
            }
        }
    }
    /*
    * 如果入力p_st_data_in是FIN包, 说明本次KAFKA的connection结束,
    * 释放本次session信息. 如果入力p_st_data_in是SYN包,
    * 说明新的KAFKA的connection开始了, 释放上次可能残留的session信息
    */
    if (p_st_packet->em_analyz_cmd == FRANALYZ_CMD_FIN
        || p_st_packet->em_analyz_cmd == FRANALYZ_CMD_SYN)
    {
        /* 销毁所有相关本次KAFKA数据 */
        __kafka_destroy_curr_info((L7session_kafka_private *)p_st_session->p_session_private);
        return APM_L7_ANALYZE_SUCC;
    }

    if (p_st_packet->ui_payload_len == 0)
    {
        return APM_L7_ANALYZE_TODO;
    }
    p_st_private = (L7session_kafka_private *)p_st_session->p_session_private;
    /* 设置解析数据 */
    i_ret = __kafka_set_analysis_data(p_st_conn, p_st_packet, p_st_private);
    if (i_ret == APM_L7_ANALYZE_TODO)
    {
        frdbg("need more data\n");
        return APM_L7_ANALYZE_TODO;
    }
    else if (i_ret == APM_L7_ANALYZE_FAIL)
    {
        frdbg("call __kafka_set_analysis_data failed\n");
        __kafka_destroy_curr_info(p_st_private);
        return APM_L7_ANALYZE_FAIL;
    }

    if (p_st_private->i_version == KAFKA_VERSION_8)
    {
        /* 解析packet */
        switch (p_st_private->p_st_fragment->i_req_resp)
        {
            case FR_CLIENT_IP_PORT: /* 解析request */
                i_ret = kafka_parse_request(p_st_conn, p_st_hash_report, p_st_private);
                if (i_ret == APM_L7_ANALYZE_FAIL)
                {
                    frdbg("Parse request failed\n");
                    __kafka_destroy_curr_info(p_st_private);
                    return APM_L7_ANALYZE_FAIL;
                }
                break;

            case FR_SRV_IP_PORT: /* 解析response */
                i_ret = kafka_parse_response(p_st_conn, p_st_hash_report, p_st_private);
                __kafka_destroy_curr_info(p_st_private);
                if (i_ret == APM_L7_ANALYZE_FAIL)
                {
                    frdbg("Parse response failed\n");
                    return APM_L7_ANALYZE_FAIL;
                }
                break;

            default: /* 类型未知 */
                i_ret = APM_L7_ANALYZE_FAIL;
                break;
        }
    }
    else
    {
        /* 解析packet */
        switch (p_st_private->p_st_fragment->i_req_resp)
        {
            case FR_CLIENT_IP_PORT: /* 解析request */
                i_ret = kafka_parse_request_7(p_st_conn, p_st_hash_report, p_st_private);
                if (i_ret == APM_L7_ANALYZE_FAIL)
                {
                    frdbg("Parse request failed\n");
                    __kafka_destroy_curr_info(p_st_private);
                    return APM_L7_ANALYZE_FAIL;
                }
                break;

            case FR_SRV_IP_PORT: /* 解析response */
                i_ret = kafka_parse_response_7(p_st_conn, p_st_hash_report, p_st_private);
                __kafka_destroy_curr_info(p_st_private);
                if (i_ret == APM_L7_ANALYZE_FAIL)
                {
                    frdbg("Parse response failed\n");
                    return APM_L7_ANALYZE_FAIL;
                }
                break;

            default: /* 类型未知 */
                i_ret = APM_L7_ANALYZE_FAIL;
                break;
        }
    }

    /* 返回 */
    return i_ret;
}

/*******************************************************************************
*  FUNC     :  私有信息销毁
*  ARGS     :  p_private_session(KAFKA的私有session数据)
*  RTN      :  void
*  NOTE     :
*******************************************************************************/
int kafka_session_destroy(void *p_private_session)
{
    L7session_kafka_private *p_st_private = (L7session_kafka_private *)p_private_session;

    /* 判断session是否为空 */
    if (p_st_private == NULL)
    {
        return FR_SUCC;
    }

    /* 销毁request数据 */
    if (p_st_private->p_st_req)
    {
        kafka_destroy_request(&(p_st_private->p_st_req));
    }

    /* 销毁response数据 */
    if (p_st_private->p_st_resp)
    {
        kafka_destroy_response(&p_st_private->p_st_resp);
    }

    /* 销毁分片数据 */
    if (p_st_private->p_st_fragment)
    {
        kafka_destroy_fragment(&p_st_private->p_st_fragment);
    }

    frfree(p_st_private);

    return FR_SUCC;
}
